package mysql

import (
	"database/sql"
	"fmt"
	db_utils "gitee.com/zy_blue7/db-convert/db-utils"
	log2 "gitee.com/zy_blue7/db-convert/db-utils/log"
	_ "github.com/go-sql-driver/mysql"
	"github.com/jmoiron/sqlx"
	"log"
	"strings"
)

const (
	Batch_Size = 10
)

type Mysql struct {
	DriverName string
	UserName   string
	Password   string
	Host       string
	Port       string
	Schema     string
	DB         db_utils.DB
	Convert    db_utils.ConvertValue
}

func (m *Mysql) GetDB() db_utils.DB {
	return m.DB
}

func (m *Mysql) GetSchema() string {
	return m.Schema
}

var _ db_utils.Databases = (*Mysql)(nil)
var _ db_utils.Tables = (*Mysql)(nil)
var _ db_utils.TableColumns = (*Mysql)(nil)
var _ db_utils.PrimaryKeys = (*Mysql)(nil)
var _ db_utils.ForeignKeys = (*Mysql)(nil)
var _ db_utils.Indexes = (*Mysql)(nil)
var _ db_utils.DataSchema = (*Mysql)(nil)
var _ db_utils.DataConvert = (*Mysql)(nil)

type DBLog struct {
	Log log2.Log
	DB  *sqlx.DB
}

func (db *DBLog) Query(query string, args ...interface{}) (*sql.Rows, error) {
	db.Log.Debug(fmt.Sprintf("Query: %s,\n Args: %+v", query, args))
	rows, err := db.DB.Query(query, args...)
	return rows, err
}

func (db *DBLog) Queryx(query string, args ...interface{}) (*sqlx.Rows, error) {
	db.Log.Debug(fmt.Sprintf("Queryx: %s,\n Args: %+v", query, args))
	rows, err := db.DB.Queryx(query, args...)
	return rows, err
}

func (db *DBLog) QueryRowx(query string, args ...interface{}) *sqlx.Row {
	db.Log.Debug(fmt.Sprintf("QueryRowx: %s,\n Args: %+v", query, args))
	rows := db.DB.QueryRowx(query, args...)
	return rows
}

func (db *DBLog) Exec(query string, args ...interface{}) (sql.Result, error) {
	db.Log.Debug(fmt.Sprintf("Exec: %s,\n Args: %+v", query, args))
	rows, err := db.DB.Exec(query, args...)
	return rows, err
}

func (db *DBLog) Select(dest interface{}, query string, args ...interface{}) error {
	db.Log.Debug(fmt.Sprintf("Select: %s,\n Args: %+v", query, args))
	err := db.DB.Select(dest, query, args...)
	return err
}

func (db *DBLog) Get(dest interface{}, query string, args ...interface{}) error {
	db.Log.Debug(fmt.Sprintf("Get: %s,\n Args: %+v", query, args))
	err := db.DB.Get(dest, query, args...)
	return err
}

var _ db_utils.DB = (*DBLog)(nil)

func NewDB(log3 log2.Log, db2 *sqlx.DB) db_utils.DB {
	return &DBLog{
		Log: log3,
		DB:  db2,
	}
}

type Option func(*Mysql)

func NewMysql(opts ...Option) *Mysql {
	mysql := &Mysql{
		DriverName: "",
		UserName:   "",
		Password:   "",
		Host:       "",
		Port:       "",
		Schema:     "",
	}
	for _, opt := range opts {
		opt(mysql)
	}
	conn := mysql.Conn()
	mysql.DB = NewDB(log2.NewLog(log2.DEBUG), conn)
	return mysql
}

func WithConvertValue(Convert db_utils.ConvertValue) Option {
	return func(mysql *Mysql) {
		mysql.Convert = Convert
	}
}

func WithHost(host string) Option {
	return func(mysql *Mysql) {
		mysql.Host = host
	}
}

func WithDriverName(DriverName string) Option {
	return func(mysql *Mysql) {
		mysql.DriverName = DriverName
	}
}

func WithUserName(UserName string) Option {
	return func(mysql *Mysql) {
		mysql.UserName = UserName
	}
}
func WithPassword(Password string) Option {
	return func(mysql *Mysql) {
		mysql.Password = Password
	}
}
func WithPort(Port string) Option {
	return func(mysql *Mysql) {
		mysql.Port = Port
	}
}
func WithSchema(Schema string) Option {
	return func(mysql *Mysql) {
		mysql.Schema = Schema
	}
}

func (m *Mysql) Conn() *sqlx.DB {
	// 连接数据库
	//db, err := sqlx.Connect("mysql", "bito:bitorobotics@tcp(127.0.0.1:33063)/mep_db_tmp?parseTime=true")
	db, err := sqlx.Connect(m.DriverName, fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true", m.UserName, m.Password, m.Host, m.Port, m.Schema))
	if err != nil {
		log.Fatalln(err)
	}
	return db
}

func (m *Mysql) GetDatabases() ([]string, error) {
	var databases []string

	// MySQL 方式
	err := m.DB.Select(&databases, "SHOW DATABASES")

	// PostgreSQL 方式
	// err := db.Select(&databases, "SELECT datname FROM pg_database WHERE datistemplate = false")

	// SQL Server 方式
	// err := db.Select(&databases, "SELECT name FROM sys.databases WHERE database_id > 4")

	if err != nil {
		return nil, fmt.Errorf("获取数据库列表失败: %w", err)
	}
	return databases, nil
}

func (m *Mysql) GetTables(schema string) ([]string, error) {
	var tables []string

	// 通用方式 (information_schema)
	// query := "SELECT table_name FROM information_schema.tables WHERE table_schema = ?"
	// err := m.DB.Select(&tables, query, schema)

	// MySQL 专用方式
	err := m.DB.Select(&tables, "SHOW TABLES")

	// PostgreSQL 专用方式
	// err := db.Select(&tables, "SELECT tablename FROM pg_tables WHERE schemaname = $1", schema)

	if err != nil {
		return nil, fmt.Errorf("获取表列表失败: %w", err)
	}
	return tables, nil
}

func (m *Mysql) GetTableColumns(schema, tableName string) ([]db_utils.ColumnInfo, error) {
	var columns []db_utils.ColumnInfo

	query := `
        SELECT
            TABLE_NAME AS table_name,
            COLUMN_NAME AS column_name,
            DATA_TYPE AS data_type,
            IS_NULLABLE AS is_nullable,
            COLUMN_DEFAULT AS column_default,
            CHARACTER_MAXIMUM_LENGTH AS character_maximum_length,
            NUMERIC_PRECISION AS numeric_precision,
            NUMERIC_SCALE AS numeric_scale,
            COLUMN_KEY AS column_key,
            EXTRA AS extra,
            COLUMN_COMMENT AS column_comment
        FROM
            information_schema.columns
        WHERE
            TABLE_SCHEMA = ? AND TABLE_NAME = ?
        ORDER BY
            TABLE_NAME,ORDINAL_POSITION
    `

	// pgsql 语法
	//	query := `
	//		SELECT
	//            a.attname AS column_name,
	//            pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type,
	//            NOT a.attnotnull AS is_nullable,
	//            pg_catalog.pg_get_expr(ad.adbin, ad.adrelid) AS column_default,
	//            CASE
	//                WHEN a.atttypmod = -1 THEN NULL
	//                WHEN t.typname IN ('varchar', 'bpchar') THEN a.atttypmod - 4
	//                ELSE NULL
	//            END AS character_maximum_length,
	//            CASE
	//                WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) >> 16
	//                ELSE NULL
	//            END AS numeric_precision,
	//            CASE
	//                WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) & 65535
	//                ELSE NULL
	//            END AS numeric_scale,
	//            d.description column_comment
	//        FROM
	//            pg_attribute a
	//        JOIN
	//            pg_class c ON a.attrelid = c.oid
	//        JOIN
	//            pg_namespace n ON c.relnamespace = n.oid
	//        JOIN
	//            pg_type t ON a.atttypid = t.oid
	//        LEFT JOIN
	//            pg_attrdef ad ON a.attrelid = ad.adrelid AND a.attnum = ad.adnum
	//        LEFT JOIN
	//            pg_description d ON a.attrelid = d.objoid AND a.attnum = d.objsubid
	//        WHERE
	//            n.nspname = ? AND c.relname =  ?
	//            AND a.attnum > 0 AND NOT a.attisdropped
	//        ORDER BY
	//            a.attnum
	//`

	err := m.DB.Select(&columns, query, schema, tableName)
	if err != nil {
		return nil, fmt.Errorf("获取表列信息失败: %w", err)
	}
	return columns, nil
}

func (m *Mysql) GetAllTableColumns(schema string) ([]db_utils.ColumnInfo, error) {
	var columns []db_utils.ColumnInfo

	query := `
         SELECT
            TABLE_NAME AS table_name,
            COLUMN_NAME AS column_name,
            DATA_TYPE AS data_type,
            IS_NULLABLE AS is_nullable,
            COLUMN_DEFAULT AS column_default,
            CHARACTER_MAXIMUM_LENGTH AS character_maximum_length,
            NUMERIC_PRECISION AS numeric_precision,
            NUMERIC_SCALE AS numeric_scale,
            COLUMN_KEY AS column_key,
            EXTRA AS extra,
            COLUMN_COMMENT AS column_comment
        FROM
            information_schema.columns
        WHERE
            TABLE_SCHEMA = ? 
  -- AND TABLE_NAME = ?
        ORDER BY
            TABLE_NAME,ORDINAL_POSITION
    `

	// pgsql 语法
	//	query := `
	//		SELECT
	//            a.attname AS column_name,
	//            pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type,
	//            NOT a.attnotnull AS is_nullable,
	//            pg_catalog.pg_get_expr(ad.adbin, ad.adrelid) AS column_default,
	//            CASE
	//                WHEN a.atttypmod = -1 THEN NULL
	//                WHEN t.typname IN ('varchar', 'bpchar') THEN a.atttypmod - 4
	//                ELSE NULL
	//            END AS character_maximum_length,
	//            CASE
	//                WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) >> 16
	//                ELSE NULL
	//            END AS numeric_precision,
	//            CASE
	//                WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) & 65535
	//                ELSE NULL
	//            END AS numeric_scale,
	//            d.description column_comment
	//        FROM
	//            pg_attribute a
	//        JOIN
	//            pg_class c ON a.attrelid = c.oid
	//        JOIN
	//            pg_namespace n ON c.relnamespace = n.oid
	//        JOIN
	//            pg_type t ON a.atttypid = t.oid
	//        LEFT JOIN
	//            pg_attrdef ad ON a.attrelid = ad.adrelid AND a.attnum = ad.adnum
	//        LEFT JOIN
	//            pg_description d ON a.attrelid = d.objoid AND a.attnum = d.objsubid
	//        WHERE
	//            n.nspname = ? AND c.relname =  ?
	//            AND a.attnum > 0 AND NOT a.attisdropped
	//        ORDER BY
	//            a.attnum
	//`

	err := m.DB.Select(&columns, query, schema)
	if err != nil {
		return nil, fmt.Errorf("获取表列信息失败: %w", err)
	}
	return columns, nil
}

func (m *Mysql) GetPrimaryKeys(schema, tableName string) ([]db_utils.PrimaryKeyInfo, error) {
	var pks []db_utils.PrimaryKeyInfo

	query := `
        SELECT 
            tc.TABLE_NAME AS table_name,
            kcu.COLUMN_NAME AS column_name,
            kcu.ORDINAL_POSITION  as key_seq
        FROM 
            information_schema.table_constraints tc
        JOIN 
            information_schema.key_column_usage kcu
            ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
            AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA
        	AND tc.TABLE_NAME = kcu.TABLE_NAME
        WHERE 
            tc.CONSTRAINT_TYPE = 'PRIMARY KEY'
            AND tc.TABLE_SCHEMA = ?
            AND tc.TABLE_NAME = ?
        ORDER BY 
            tc.TABLE_NAME,kcu.ORDINAL_POSITION
    `

	err := m.DB.Select(&pks, query, schema, tableName)
	if err != nil {
		return nil, fmt.Errorf("获取主键信息失败: %w", err)
	}
	return pks, nil
}

func (m *Mysql) GetAllPrimaryKeys(schema string) ([]db_utils.PrimaryKeyInfo, error) {
	var pks []db_utils.PrimaryKeyInfo

	query := `
        SELECT 
            tc.TABLE_NAME AS table_name,
            kcu.COLUMN_NAME AS column_name,
            kcu.ORDINAL_POSITION  as key_seq
        FROM 
            information_schema.table_constraints tc
        JOIN 
            information_schema.key_column_usage kcu
            ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
            AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA
        	AND tc.TABLE_NAME = kcu.TABLE_NAME
        WHERE 
            tc.CONSTRAINT_TYPE = 'PRIMARY KEY'
            AND tc.TABLE_SCHEMA = ?
   --         AND tc.TABLE_NAME = ?
        ORDER BY 
            tc.TABLE_NAME,kcu.ORDINAL_POSITION
    `

	err := m.DB.Select(&pks, query, schema)
	if err != nil {
		return nil, fmt.Errorf("获取主键信息失败: %w", err)
	}
	return pks, nil
}

func (m *Mysql) GetForeignKeys(schema, tableName string) ([]db_utils.ForeignKeyInfo, error) {
	var fks []db_utils.ForeignKeyInfo

	// mysql
	query := `
       SELECT
        distinct   tc.CONSTRAINT_NAME AS constraint_name,
           kcu.COLUMN_NAME AS column_name,
           kcu.TABLE_NAME table_name,
           kcu.REFERENCED_TABLE_NAME AS foreign_table_name,
           kcu.REFERENCED_COLUMN_NAME AS foreign_column_name,
           rc.UPDATE_RULE AS update_rule,
           rc.DELETE_RULE AS delete_rule
       FROM
           information_schema.table_constraints AS tc
       JOIN
           information_schema.key_column_usage AS kcu
           ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
           AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA
       JOIN
           information_schema.REFERENTIAL_CONSTRAINTS AS ccu
           ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
       JOIN
           information_schema.referential_constraints AS rc
           ON rc.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
       WHERE
           tc.CONSTRAINT_TYPE = 'FOREIGN KEY'
           AND tc.TABLE_SCHEMA = ?
           AND tc.TABLE_NAME = ?
	   ORDER BY  tc.TABLE_NAME 
    `

	// pgsql 语法
	//query := `
	//SELECT
	//        tc.constraint_name,
	//        kcu.column_name,
	//        kcu.table_name,
	//        ccu.table_name AS foreign_table_name,
	//        ccu.column_name AS foreign_column_name,
	//        rc.update_rule,
	//        rc.delete_rule
	//    FROM
	//        information_schema.table_constraints tc
	//    JOIN
	//        information_schema.key_column_usage kcu
	//        ON tc.constraint_name = kcu.constraint_name
	//        AND tc.table_schema = kcu.table_schema
	//    JOIN
	//        information_schema.constraint_column_usage ccu
	//        ON ccu.constraint_name = tc.constraint_name
	//    JOIN
	//        information_schema.referential_constraints rc
	//        ON rc.constraint_name = tc.constraint_name
	//    WHERE
	//        tc.constraint_type = 'FOREIGN KEY'
	//        AND tc.table_schema = ?
	//        AND tc.table_name = ?
	//`

	err := m.DB.Select(&fks, query, schema, tableName)
	if err != nil {
		return nil, fmt.Errorf("获取外键信息失败: %w", err)
	}
	return fks, nil
}

func (m *Mysql) GetAllForeignKeys(schema string) ([]db_utils.ForeignKeyInfo, error) {
	var fks []db_utils.ForeignKeyInfo

	// mysql
	query := `
       SELECT
           tc.CONSTRAINT_NAME AS constraint_name,
           kcu.COLUMN_NAME AS column_name,
           kcu.TABLE_NAME AS table_name,
           kcu.REFERENCED_TABLE_NAME AS foreign_table_name,
           kcu.REFERENCED_COLUMN_NAME AS foreign_column_name,
           rc.UPDATE_RULE AS update_rule,
           rc.DELETE_RULE AS delete_rule
       FROM
           information_schema.table_constraints AS tc
       JOIN
           information_schema.key_column_usage AS kcu
           ON tc.CONSTRAINT_NAME = kcu.CONSTRAINT_NAME
           AND tc.TABLE_SCHEMA = kcu.TABLE_SCHEMA
       JOIN
           information_schema.REFERENTIAL_CONSTRAINTS AS ccu
           ON ccu.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
       JOIN
           information_schema.referential_constraints AS rc
           ON rc.CONSTRAINT_NAME = tc.CONSTRAINT_NAME
       WHERE
           tc.CONSTRAINT_TYPE = 'FOREIGN KEY'
           AND tc.TABLE_SCHEMA = ?
    --       AND tc.TABLE_NAME = ?
	   ORDER BY  tc.TABLE_NAME 
    `

	// pgsql 语法
	//query := `
	//SELECT
	//        tc.constraint_name,
	//        kcu.column_name,
	//        kcu.table_name,
	//        ccu.table_name AS foreign_table_name,
	//        ccu.column_name AS foreign_column_name,
	//        rc.update_rule,
	//        rc.delete_rule
	//    FROM
	//        information_schema.table_constraints tc
	//    JOIN
	//        information_schema.key_column_usage kcu
	//        ON tc.constraint_name = kcu.constraint_name
	//        AND tc.table_schema = kcu.table_schema
	//    JOIN
	//        information_schema.constraint_column_usage ccu
	//        ON ccu.constraint_name = tc.constraint_name
	//    JOIN
	//        information_schema.referential_constraints rc
	//        ON rc.constraint_name = tc.constraint_name
	//    WHERE
	//        tc.constraint_type = 'FOREIGN KEY'
	//        AND tc.table_schema = ?
	//        AND tc.table_name = ?
	//`

	err := m.DB.Select(&fks, query, schema)
	if err != nil {
		return nil, fmt.Errorf("获取外键信息失败: %w", err)
	}
	return fks, nil
}
func (m *Mysql) GetIndexes(schema, tableName string) ([]db_utils.IndexInfo, error) {
	var indexes []db_utils.IndexInfo

	query := `
        SELECT 
            INDEX_NAME AS index_name,
            COLUMN_NAME AS column_name,
            NON_UNIQUE AS non_unique,
            INDEX_TYPE AS index_type
        FROM 
            information_schema.statistics
        WHERE 
            TABLE_SCHEMA = ?
            AND TABLE_NAME = ?
        ORDER BY 
            INDEX_NAME, SEQ_IN_INDEX
    `

	// pgsql 语法
	//query := `
	//SELECT distinct i.relname                                   AS indexname,
	//				a.attname                                   AS column_name,
	//				case when idx.indisunique then 1 else 0 end AS non_unique,
	//				am.amname                                      index_type,
	//				pg_get_indexdef(idx.indexrelid)             AS index_def
	//FROM pg_class t,
	// 	pg_class i,
	//	 pg_am am,
	// 	pg_index idx,
	// 	pg_attribute a,
	//	 pg_namespace n
	//WHERE t.oid = idx.indrelid
	//	AND i.oid = idx.indexrelid
	//	AND a.attrelid = t.oid
	//	AND a.attnum = ANY (idx.indkey)
	//	AND t.relnamespace = n.oid
	//	AND n.nspname = ?
	// 	AND t.relname = ?
	//ORDER BY i.relname
	//`

	err := m.DB.Select(&indexes, query, schema, tableName)
	if err != nil {
		return nil, fmt.Errorf("获取索引信息失败: %w", err)
	}
	return indexes, nil
}
func (m *Mysql) GetAllIndexes(schema string) ([]db_utils.IndexInfo, error) {
	var indexes []db_utils.IndexInfo

	query := `
        SELECT 
			TABLE_NAME AS table_name,
            INDEX_NAME AS index_name,
            COLUMN_NAME AS column_name,
            NON_UNIQUE AS non_unique,
            INDEX_TYPE AS index_type
        FROM 
            information_schema.statistics
        WHERE 
            TABLE_SCHEMA = ?
   --          AND TABLE_NAME = ?
        ORDER BY 
            INDEX_NAME, SEQ_IN_INDEX
    `

	// pgsql 语法
	//query := `
	//SELECT distinct i.relname                                   AS indexname,
	//				a.attname                                   AS column_name,
	//				case when idx.indisunique then 1 else 0 end AS non_unique,
	//				am.amname                                      index_type,
	//				pg_get_indexdef(idx.indexrelid)             AS index_def
	//FROM pg_class t,
	// 	pg_class i,
	//	 pg_am am,
	// 	pg_index idx,
	// 	pg_attribute a,
	//	 pg_namespace n
	//WHERE t.oid = idx.indrelid
	//	AND i.oid = idx.indexrelid
	//	AND a.attrelid = t.oid
	//	AND a.attnum = ANY (idx.indkey)
	//	AND t.relnamespace = n.oid
	//	AND n.nspname = ?
	// 	AND t.relname = ?
	//ORDER BY i.relname
	//`

	err := m.DB.Select(&indexes, query, schema)
	if err != nil {
		return nil, fmt.Errorf("获取索引信息失败: %w", err)
	}
	return indexes, nil
}

// ConvertData 转换数据核心方法
// m 源数据源，destDataSchema 目的数据源，config 转换配置
func (m *Mysql) ConvertData(destDataSchema db_utils.DataSchema, config db_utils.ConvertConfig) error {
	logs := log2.NewLog(log2.DEBUG)
	selectSql := "SELECT %s FROM %s"
	for sourceTable, destTableValue := range config.Result {
		for destTable, columns := range destTableValue {
			sourceTableColumns, err := m.GetTableColumns(m.Schema, sourceTable)
			destTableColumns, err := destDataSchema.GetTableColumns(destDataSchema.GetSchema(), destTable)
			if err != nil {
				log.Fatalln(err)
			}
			if err != nil {
				panic(err)
			}
			sourceColumnNames := make([]string, len(sourceTableColumns))
			for i, column := range sourceTableColumns {
				sourceColumnNames[i] = column.ColumnName
				logs.Debug(fmt.Sprintf("%v", column))
			}
			destColumnNames := make([]string, len(destTableColumns))
			for i, column := range destTableColumns {
				destColumnNames[i] = column.ColumnName
				logs.Debug(fmt.Sprintf("%v", column))
			}
			// dest-column: source-column
			destSourceColMap := make(map[string]string, len(columns))
			for k, v := range columns {
				destSourceColMap[v] = k
			}
			// insert sql   INSERT INTO %s (%s) VALUES
			insertPrefixSql := buildInsertPrefixSQL(destTable, destColumnNames)
			fmt.Println(insertPrefixSql)
			rows, err := m.DB.Query(fmt.Sprintf(selectSql, strings.Join(sourceColumnNames, ","), sourceTable))
			if err != nil {
				logs.Error(fmt.Sprintf("%v", err))
				continue
			}

			var num int = 0
			//valuesTmp := make([]interface{}, Batch_Size*len(columns))
			valuesTmp := []interface{}{}
			placeholderOne := []string{}
			placeholder := []string{}
			for i := 0; i < len(destColumnNames); i++ {
				placeholderOne = append(placeholderOne, "?")
			}
			// (?,?,?,?)
			placeholderOneStr := fmt.Sprintf("( %s )", strings.Join(placeholderOne, ","))
			for rows.Next() {
				if num > Batch_Size {
					//valueStr
					placeholderStr := fmt.Sprintf("  %s  ", strings.Join(placeholder, ","))
					placeholder = placeholder[0:0]
					batchInsertSql := fmt.Sprintf("%s %s", insertPrefixSql, placeholderStr)
					fmt.Println(batchInsertSql)
					result, err := destDataSchema.GetDB().Exec(batchInsertSql, valuesTmp...)
					//myLog.Debug(fmt.Sprintf("Exec: %s,\n Args: %+v", batchInsertSql, valuesTmp))
					//myLog.Debug(fmt.Sprintf("Exec: %s,\n Args: %#v", batchInsertSql, valuesTmp))
					//myLog.Debug(fmt.Sprintf("Exec: %s,\n Args: %v", batchInsertSql, valuesTmp))
					if err != nil {
						logs.Error(fmt.Sprintf("err : %v \n Exec sql %s \n, Args: %v ", err, batchInsertSql, valuesTmp))
						log.Fatal(err)
					}
					lastInsertId, err := result.LastInsertId()
					rowsAffected, err := result.RowsAffected()
					log.Println(fmt.Sprintf("lastInsertId %d ,rowsAffected %d", lastInsertId, rowsAffected))
					valuesTmp = valuesTmp[0:0]
					num = 0
				}
				// (?,?,?,?),(?,?,?,?),(?,?,?,?)
				placeholder = append(placeholder, placeholderOneStr)

				// 获取列类型信息
				// /Users/blue7/go/pkg/mod/github.com/go-sql-driver/mysql@v1.9.2/fields.go
				// todo 每个连接对象都有一个这样的实现，里面列出了所有的mysql的数据类型
				colTypes, err := rows.ColumnTypes()
				if err != nil {
					log.Fatal("获取列类型失败:", err)
				}
				// 创建动态值容器
				values := make([]interface{}, len(sourceTableColumns))
				for i := range values {
					values[i] = new(interface{})
				}
				if err := rows.Scan(values...); err != nil {
					fmt.Println("填充值失败。。。")
					panic(err)
				}
				valueMap := make(map[string]interface{}, len(values))
				// 转换值为可插入格式
				insertValues := make([]interface{}, len(values))
				for i, v := range values {
					val := *(v.(*interface{}))
					insertValues[i] = m.Convert.ConvertValue(val, colTypes[i].DatabaseTypeName())
					valueMap[sourceColumnNames[i]] = insertValues[i]
				}
				for _, columnName := range destColumnNames {
					// 根据 column映射，得到source column 名称
					sourceColumn := destSourceColMap[columnName]
					v := valueMap[sourceColumn]
					if v != nil {
						valuesTmp = append(valuesTmp, v)
						continue
					}
					valuesTmp = append(valuesTmp, nil)
				}
				//fmt.Println(insertValues)
				//for _, value := range insertValues {
				//	valuesTmp = append(valuesTmp, value)
				//}
				num++
			}
		}
	}

	return nil
}

func buildInsertPrefixSQL(table string, columns []string) string {

	return fmt.Sprintf("INSERT INTO %s (%s) VALUES ",
		table,
		strings.Join(columns, ","))
}
